package indi.zhifa.study2025.flinkcdc;


import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import indi.zhifa.study2025.commondata.GenItemEntity;
import indi.zhifa.study2025.flinkcdc.binlog.BinlogInfo;
import indi.zhifa.study2025.flinkcdc.elsearch.ElSearchSink;
import indi.zhifa.study2025.flinkcdc.elsearch.command.EElCommand;
import indi.zhifa.study2025.flinkcdc.elsearch.command.ElCommand;
import indi.zhifa.study2025.flinkcdc.redis.SpringDataRedisSink;
import indi.zhifa.study2025.flinkcdc.redis.command.RedisSinkCommand;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;


public class FlinkMain {

    public static void main(String[] args) throws Exception {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("192.168.0.64")
                .port(3307)
                .databaseList("shop-center") // set captured database
                .tableList("shop-center.item") // set captured table
                .username("app")
                .password("study@2025")
                .serverTimeZone("Asia/Shanghai")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.latest())
                .includeSchemaChanges(true)
                .build();

//        FlinkJedisPoolConfig jedisConfig = new FlinkJedisPoolConfig.Builder()
//                .setHost("192.168.0.64") // 替换为 Redis 主机
//                .setPort(6379) // Redis 端口
//                .setPassword("ilv0404@1314") // 如果有密码，设置密码
//                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        DataStream<BinlogInfo> mysqlStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source")
//                .map(str->{
//                    BinlogInfo res =JSONObject.parseObject(str, BinlogInfo.class);
//                    return res;
//                    }
//                 ).filter(bi->bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d"));
//
//        mysqlStream.addSink(new RedisSink(jedisConfig,new RedisItemMapper()));

        DataStream<RedisSinkCommand<GenItemEntity>> newMysqlStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source to redis")
                        .map(str->JSONObject.parseObject(str, new TypeReference<BinlogInfo<GenItemEntity>>() {
                        }), TypeInformation.of(new TypeHint<BinlogInfo<GenItemEntity>>() {}))
                .filter(bi->bi.getSource().getTable().equals("item") &&  (bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d")))
                .map(bi->{
                    String op = bi.getOp();
                    GenItemEntity itemEntity = bi.getAfter();
                    String key = "item:"+itemEntity.getId();
                    switch (op){
                        case "c","u"->{
                            RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
                            redisSinkCommand.initSet(key,itemEntity);
                            return redisSinkCommand;
                        }
                        case "d" ->{
                            RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
                            redisSinkCommand.initDel(key);
                            return redisSinkCommand;
                        }
                        default -> {
                            RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
                            redisSinkCommand.initDel(key);
                            return redisSinkCommand;
                        }
                    }
                },TypeInformation.of(new TypeHint<RedisSinkCommand<GenItemEntity>>() {}));

        newMysqlStream.sinkTo(new SpringDataRedisSink<GenItemEntity>());

        DataStream<ElCommand<GenItemEntity>> mySqlToElStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source to el")
                .map(str->JSONObject.parseObject(str, new TypeReference<BinlogInfo<GenItemEntity>>() {})
                        , TypeInformation.of(new TypeHint<BinlogInfo<GenItemEntity>>() {}))
        .filter(bi->bi.getSource().getTable().equals("item") &&  (bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d")))
                .map(bi->{
                    ElCommand elCommand = new ElCommand();
                    GenItemEntity itemEntity = bi.getAfter();
                    elCommand.setId(itemEntity.getId().toString());
                    elCommand.setEntity(itemEntity);
                    elCommand.setIndex("item_npc");
                    String op = bi.getOp();
                    switch (op){
                        case "c"->elCommand.setCommand(EElCommand.CREATE);
                        case "u"->elCommand.setCommand(EElCommand.UPDATE);
                        case "d"->elCommand.setCommand(EElCommand.DELETE);
                    }
                    return elCommand;
                },TypeInformation.of(new TypeHint<ElCommand<GenItemEntity>>() {}));
        mySqlToElStream.sinkTo(new ElSearchSink());

        env.execute();

    }


}
